This notebook executes all phconvert notebooks to automate testing.
import os
from pathlib import Path
import nbformat
from nbconvert.preprocessors import ExecutePreprocessor
from IPython.display import display, FileLink
def run_notebook(in_filepath, run_path=None, out_dir=None, out_suffix=''):
"""Runs the notebook `notebook_name` (file name with no extension).
This function executes notebook with name `notebook_name` (no extension)
and saves the fully executed notebook in a new file appending "-out"
to the original file name.
It also displays links to the original and executed notebooks.
"""
if not in_filepath.is_file():
raise IOError('File "%s" not found.' % in_filepath)
in_filepath = in_filepath.resolve()
if run_path is None:
run_path = str(in_filepath.parent)
if out_dir is None:
out_dir = in_filepath.parent
else:
out_dir = Path(out_dir).resolve()
out_filepath = Path(out_dir, "%s%s.ipynb" % (in_filepath.stem, out_suffix))
nb = nbformat.read(in_filepath.open(), as_version=4)
ep = ExecutePreprocessor(timeout = 3600)
try:
out = ep.preprocess(nb, {'metadata': {'path': run_path}})
except Exception:
msg = 'Error executing the notebook "%s".\n\n' % in_filepath
msg += 'See notebook "%s" for the traceback.' % out_filepath
print(msg)
raise
finally:
nbformat.write(nb, out_filepath.open(mode='wt'))
## Monkey patch needed until nbconvert > 4.0.0 is released
## https://github.com/jupyter/nbconvert/releases
from nbformat.v4 import output_from_msg
from nbconvert.preprocessors.execute import CellExecutionError
from textwrap import dedent
def preprocess_cell(self, cell, resources, cell_index):
"""
Apply a transformation on each code cell. See base.py for details.
"""
if cell.cell_type != 'code':
return cell, resources
outputs = self.run_cell(cell)
cell.outputs = outputs
if not self.allow_errors:
for out in outputs:
if out.output_type == 'error':
pattern = """\
An error occurred while executing the following cell:
------------------
{cell.source}
------------------
{out.ename}: {out.evalue}
"""
msg = dedent(pattern).format(out=out, cell=cell)
raise CellExecutionError(msg)
return cell, resources
def run_cell(self, cell):
msg_id = self.kc.execute(cell.source)
self.log.debug("Executing cell:\n%s", cell.source)
# wait for finish, with timeout
while True:
try:
msg = self.kc.shell_channel.get_msg(timeout=self.timeout)
except Empty:
self.log.error("""Timeout waiting for execute reply (%is).
If your cell should take longer than this, you can increase the timeout with:
c.ExecutePreprocessor.timeout = SECONDS
in jupyter_nbconvert_config.py
""" % self.timeout)
if self.interrupt_on_timeout:
self.log.error("Interrupting kernel")
self.km.interrupt_kernel()
break
else:
try:
exception = TimeoutError
except NameError:
exception = RuntimeError
raise exception("Cell execution timed out, see log"
" for details.")
if msg['parent_header'].get('msg_id') == msg_id:
break
else:
# not our reply
continue
outs = []
while True:
try:
msg = self.kc.iopub_channel.get_msg(timeout=self.timeout)
except Empty:
self.log.warn("Timeout waiting for IOPub output")
break
if msg['parent_header'].get('msg_id') != msg_id:
# not an output from our execution
continue
msg_type = msg['msg_type']
self.log.debug("output: %s", msg_type)
content = msg['content']
# set the prompt number for the input and the output
if 'execution_count' in content:
cell['execution_count'] = content['execution_count']
if msg_type == 'status':
if content['execution_state'] == 'idle':
break
else:
continue
elif msg_type == 'execute_input':
continue
elif msg_type == 'clear_output':
outs = []
continue
elif msg_type.startswith('comm'):
continue
try:
out = output_from_msg(msg)
except ValueError:
self.log.error("unhandled iopub msg: " + msg_type)
else:
outs.append(out)
return outs
ExecutePreprocessor.preprocess_cell = preprocess_cell
ExecutePreprocessor.run_cell = run_cell
%pwd
pathlist = list(f for f in Path('../notebooks').glob('*.ipynb') if not f.name.startswith('_'))
pathlist
for nbpath in pathlist:
print(nbpath.stem)
run_notebook(nbpath, out_dir=Path(nbpath.parent, 'out'))